package com.yesep.netty;

import com.yesep.business.BusinessHandlerThread;
import com.yesep.constant.Constants;
import com.yesep.downstream.DownStreamDataThread;
import com.yesep.upstream.UpStreamDataTask;
import com.yesep.upstream.UpStreamDataThread;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.util.concurrent.TimeUnit;


/**
 * @program: gateway
 * @description: netty服务
 * @author: quliang
 * @create: 2019-07-04 11:46
 **/
@Component
public class NettyServer {

    private final static EventLoopGroup bossGroup = new NioEventLoopGroup(1);

    private final static EventLoopGroup workerGroup = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors() * 2);

    //下行线程  回复下位机
    private DownStreamDataThread downStreamDataThread;

    //上行线程
    private UpStreamDataThread upStreamDataThread;

    //业务线程
    private BusinessHandlerThread businessHandlerThread;

    //启动程序
    public void start(int port) {
//        this.downStreamDataThread = new DownStreamDataThread("DownStreamDataThread");
//        this.upStreamDataThread = new UpStreamDataThread("UpStreamDataThread");
        this.businessHandlerThread = new BusinessHandlerThread("BusinessHandlerThread");

        System.out.println("port:" + port);
        try {
            ServerBootstrap sbs = new ServerBootstrap()
                    .group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .localAddress(new InetSocketAddress(port))
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch)
                                throws Exception {
                            ByteBuf delimiter = Unpooled.copiedBuffer(Constants.PACKAGE_TAILED_RN_STR.getBytes());
                            ch.pipeline().addLast(
                                    new IdleStateHandler(120, 0, 0,
                                            TimeUnit.SECONDS)); // 链路空闲状态触发器
                            ch.pipeline().addLast("frameDecoder", new DelimiterBasedFrameDecoder(Integer.MAX_VALUE, delimiter));
                            ch.pipeline().addLast(new StringDecoder());
                            ch.pipeline().addLast("encoder", new StringEncoder());
                            ch.pipeline().addLast(new NettyServerHandler());
//                            ch.pipeline().addLast(new CustomMessageDecoder());
//                            ch.pipeline().addLast(new CustomMessageEncoder());
                        }
                    }).option(ChannelOption.SO_BACKLOG, 1024)
                    .option(ChannelOption.SO_SNDBUF, 1048576)
                    .option(ChannelOption.SO_RCVBUF, 1048576)
                    .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                    .childOption(ChannelOption.SO_KEEPALIVE, true);
            // 绑定端口，开始接收进来的连接
            ChannelFuture future = sbs.bind(port).sync();
            //启动成功
            if (future.isSuccess()) {
                System.out.println("启动完毕");
//                this.upStreamDataThread.init();
//                this.downStreamDataThread.init();
                this.businessHandlerThread.init();
                System.out.println("Started All ThreadPool");
//                logger.info("[GatewayServer]服务启动成功，端口号：" + this.port);
            }

            System.out.println("Server start listen at " + port);
            future.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}
